package com.bw.dws;

import com.alibaba.fastjson.JSON;
import com.bw.bean.UserProfileTag;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.phoenix.jdbc.PhoenixDriver;
import redis.clients.jedis.Jedis;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.Timestamp;

/**
 * 标签结果Sink（Phoenix+Redis+ClickHouse）
 * 文档参考：🔶1-260（输出标签属性）、🔶1-265（上线记录需导出数据）
 */
public class ProfileSink implements SinkFunction<UserProfileTag> {
    // 1. Phoenix连接信息（需替换为实际地址）
    private static final String PHOENIX_URL = "jdbc:phoenix:zk01,zk02,zk03:2181:/hbase";
    // 2. Redis连接信息
    private static final String REDIS_HOST = "redis01";
    private static final int REDIS_PORT = 6379;
    // 3. ClickHouse连接信息
    private static final String CLICKHOUSE_URL = "jdbc:clickhouse://ck01:8123/user_profile_db";
    private static final String CLICKHOUSE_USER = "default";
    private static final String CLICKHOUSE_PWD = "";

    private Connection phoenixConn;
    private PreparedStatement phoenixStmt;
    private Jedis jedis;
    private Connection ckConn;
    private PreparedStatement ckStmt;

    // 初始化连接
    public void open(org.apache.flink.configuration.Configuration parameters) throws Exception {
        // Phoenix连接
        Class.forName(PhoenixDriver.class.getName());
        phoenixConn = DriverManager.getConnection(PHOENIX_URL);
        String phoenixSql = "UPSERT INTO USER_PROFILE_TAG " +
                "(USER_ID, AGE_GROUP, GENDER_TAG, HEIGHT, WEIGHT, BIRTH_ERA, CONSTELLATION, UPDATE_TIME) " +
                "VALUES (?, ?, ?, ?, ?, ?, ?, ?)";
        phoenixStmt = phoenixConn.prepareStatement(phoenixSql);

        // Redis连接
        jedis = new Jedis(REDIS_HOST, REDIS_PORT);
        jedis.auth("123456"); // 若有密码

        // ClickHouse连接
        ckConn = DriverManager.getConnection(CLICKHOUSE_URL, CLICKHOUSE_USER, CLICKHOUSE_PWD);
        String ckSql = "INSERT INTO user_profile_tag " +
                "(user_id, age_group, gender_tag, height, weight, birth_era, constellation, update_time) " +
                "VALUES (?, ?, ?, ?, ?, ?, ?, ?)";
        ckStmt = ckConn.prepareStatement(ckSql);
    }

    // 写入数据
    @Override
    public void invoke(UserProfileTag tag, SinkFunction.Context context) throws Exception {
        String userId = tag.getUserId();
        Timestamp updateTime = tag.getUpdateTime();

        // 1. 写入Phoenix（HBase）
        phoenixStmt.setString(1, userId);
        phoenixStmt.setString(2, tag.getAgeGroup());
        phoenixStmt.setString(3, tag.getGenderTag());
        phoenixStmt.setObject(4, tag.getHeight());
        phoenixStmt.setObject(5, tag.getWeight());
        phoenixStmt.setString(6, tag.getBirthEra());
        phoenixStmt.setString(7, tag.getConstellation());
        phoenixStmt.setTimestamp(8, updateTime);
        phoenixStmt.executeUpdate();
        phoenixConn.commit(); // Phoenix需手动提交

        // 2. 写入Redis（缓存，key=user:profile:{userId}，value=JSON）
        String redisKey = "user:profile:" + userId;
        String redisValue = JSON.toJSONString(tag);
        jedis.set(redisKey, redisValue);
        jedis.expire(redisKey, 86400 * 30); // 30天过期

        // 3. 写入ClickHouse（分析查询）
        ckStmt.setString(1, userId);
        ckStmt.setString(2, tag.getAgeGroup());
        ckStmt.setString(3, tag.getGenderTag());
        ckStmt.setObject(4, tag.getHeight());
        ckStmt.setObject(5, tag.getWeight());
        ckStmt.setString(6, tag.getBirthEra());
        ckStmt.setString(7, tag.getConstellation());
        ckStmt.setTimestamp(8, updateTime);
        ckStmt.executeUpdate();
    }

    // 关闭连接
    public void close() throws Exception {
        if (phoenixStmt != null) phoenixStmt.close();
        if (phoenixConn != null) phoenixConn.close();
        if (jedis != null) jedis.close();
        if (ckStmt != null) ckStmt.close();
        if (ckConn != null) ckConn.close();
    }
}